package com.yjd.job.mns;

import com.alibaba.edas.schedulerX.ProcessResult;
import com.alibaba.edas.schedulerX.ScxSimpleJobContext;
import com.alibaba.edas.schedulerX.ScxSimpleJobProcessor;
import com.yjd.comm.base.concurrent.task.DefaultTaskDispose;
import com.yjd.comm.base.concurrent.task.ITaskHandler;
import com.yjd.comm.base.concurrent.task.ITaskItem;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.comm.order.service.IOrderService;
import com.yjd.comm.util.FrameUtil;
import com.yjd.job.order.OrderOverdueCheckJob;
import com.yjd.msg.comm.enums.MsgStatusEnum;
import com.yjd.msg.comm.model.MsgInfoModel;
import com.yjd.msg.comm.service.IMsgInfoHistoryService;
import com.yjd.msg.comm.service.IMsgInfoService;
import org.apache.log4j.Logger;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 *  消息补偿发送
 */
public class MnsMsgFailJob implements ScxSimpleJobProcessor {
    Logger logger = Logger.getLogger(MnsMsgFailJob.class);

    @Resource
    private IMsgInfoService msgInfoService;

    @Override
    public ProcessResult process(ScxSimpleJobContext scxSimpleJobContext) {
        List<MsgInfoModel> msgList = null;
        boolean flag = true;
        try {
            do {
                msgList = msgInfoService.getModelList(FrameUtil.newHashMap("next_time", FrameUtil.getTime(), "maxCount", true, "send_status", true), 0, 100, DbWREnums.READ);
                if (msgList == null || msgList.isEmpty()) {
                    break;
                }

                List<MsgModelItem> taskItems = new ArrayList<MsgModelItem>();
                for (MsgInfoModel msgInfoModel : msgList) {
                    taskItems.add(new MsgModelItem(msgInfoModel));
                }

                long starttime = System.currentTimeMillis();
                new DefaultTaskDispose(true, 5).taskDipose(taskItems,
                        new ITaskHandler<MsgInfoModel>() {
                            @Override
                            public void dispose(ITaskItem<MsgInfoModel> taskItems) {
                                if (taskItems == null|| taskItems.getItem() == null) {
                                    return;
                                }
                                try {
                                    msgInfoService.sendMessage(taskItems.getItem().getMsg_id());
                                } catch (Exception e) {
                                    logger.error("异常信息是：" + e.getMessage(), e);
                                }
                            }
                        });
                logger.info("耗时（ms）：" + (System.currentTimeMillis() - starttime) + "\t处理消息：" + (msgList.size()) + "条");
            } while (true);

        }catch (Exception e){
            logger.error("消息补偿job执行异常，异常信息:"+e);
            flag = false;
        }
        //将所有处理的数据移到历史表中
        try {
            this.moveMsgToHistory();
        } catch (Exception e) {
            logger.error("消息历史信息清理异常，异常信息:"+e);
            flag = false;
        }
        return new ProcessResult(flag);
    }

    public class MsgModelItem implements ITaskItem<MsgInfoModel> {
        private MsgInfoModel msgInfoModel;

        public MsgModelItem(MsgInfoModel msgInfoModel) {
            this.msgInfoModel = msgInfoModel;
        }

        @Override
        public MsgInfoModel getItem() {
            return msgInfoModel;
        }
    }

    private void moveMsgToHistory() throws Exception {
        while(true){
            Map<Object, Object> paramMap = new HashMap<Object, Object>();
            paramMap.put("moveHistory", "0");
            paramMap.put("status", MsgStatusEnum.CONSUME_SUCCESS.getCode());
            paramMap.put("sort", " msg_id desc ");
            paramMap.put("rowstartindex", 0);
            paramMap.put("rowendindex", 100);
            List<MsgInfoModel> list = this.msgInfoService.getModelList(paramMap,DbWREnums.WRITE);
            if (list != null && list.size()>=2) {
                list.remove(0);
                this.msgInfoService.moveMsgHistory(FrameUtil.getBeanFieldValues(list, "msg_id"));
            }else{
                break;
            }
        }
    }
}
